package mylab.utils.disruptor.event;

import com.google.common.collect.Lists;
import com.lmax.disruptor.EventTranslatorTwoArg;
import com.lmax.disruptor.ExceptionHandler;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.EventHandlerGroup;
import com.lmax.disruptor.util.DaemonThreadFactory;
import mylab.utils.common.ThreadUtil;

import java.util.List;

/**
 * File Name             :  DisruptorEventService
 *
 * @author :  sylar
 * Create                :  2019-10-21
 */
@SuppressWarnings("deprecation")
public class DisruptorEventService implements EventService {
    private static final int BUFFER_SIZE = 1024;
    private static final EventTranslatorTwoArg<DisruptorEvent, Object, Object> TRANSLATOR = (event, sequence, arg0, arg1) -> {
        event.setSource(arg0);
        event.setContent(arg1);
    };

    protected List<EventListener> listeners = Lists.newArrayList();

    protected Disruptor<DisruptorEvent> disruptor = new Disruptor<>(
            DisruptorEvent::new,
            BUFFER_SIZE,
            DaemonThreadFactory.INSTANCE
    );

    public DisruptorEventService() {
        disruptor.setDefaultExceptionHandler(new ExceptionHandler<DisruptorEvent>() {
            @Override
            public void handleEventException(Throwable throwable, long sequence, DisruptorEvent event) {
                System.err.println(String.format("###handleEventException###\nevent:%s\nerror:%s", event, throwable));
            }

            @Override
            public void handleOnStartException(Throwable ex) {
                System.err.println(String.format("###handleOnStartException###\nerror:%s", ex));
            }

            @Override
            public void handleOnShutdownException(Throwable ex) {
                System.err.println(String.format("###handleOnShutdownException###\nerror:%s", ex));
            }
        });


        EventHandlerGroup<DisruptorEvent> group;

        /**
         * handleEventsWith，所有消费者
         * 对于一条消息，handleEventsWith方法返回的EventHandlerGroup，Group中的每个消费者都会对消息进行消费，各个消费者之间不存在竞争（类似发布订阅）
         */
        group = disruptor.handleEventsWith((event, sequence, endOfBatch) -> onDisruptorEvent(event));

        /**
         * 对于一条消息，handleEventsWithWorkerPool方法返回的EventHandlerGroup，Group的消费者对于同一条消息不重复消费（类似点对点）
         */
//        group = disruptor.handleEventsWithWorkerPool(this::onDisruptorEvent);

    }

    @Override
    public void dispose() {
        stop();
    }

    @Override
    public void regist(EventListener listener) {
        if (!listeners.contains(listener)) {
            listeners.add(listener);
        }

        start();
    }

    @Override
    public void unregist(EventListener listener) {
        listeners.remove(listener);
        if (listeners.isEmpty()) {
            stop();
        }
    }

    @Override
    public void postEvent(Event<?, ?> event) {
        disruptor.publishEvent(TRANSLATOR, event.getSource(), event.getContent());
    }

    private void start() {
        disruptor.start();
    }

    private void stop() {
        disruptor.shutdown();
    }

    private void onDisruptorEvent(DisruptorEvent event) {
        listeners.forEach(listener -> ThreadUtil.postRunnable(() -> listener.onEvent(event)));
    }
}
